package blue.mqtt4j.internal.client.impl;

import blue.mqtt4j.client.MqttMessage;
import blue.mqtt4j.client.ProducerListener;
import blue.mqtt4j.client.ProducerResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;

/**
 * @author Jin Zheng
 * @since 2020-10-10
 */
public class SyncProducerListener implements ProducerListener
{
	private static Logger logger = LoggerFactory.getLogger(SyncProducerListener.class);

	private final CountDownLatch latch;
	private final BlockingQueue<ProducerResult> results;
	private final int index;

	public SyncProducerListener(CountDownLatch latch, BlockingQueue<ProducerResult> results, int index)
	{
		this.latch = latch;
		this.results = results;
		this.index = index;
	}

	@Override
	public void onComplete(MqttMessage message, Throwable cause)
	{
		if (cause != null)
		{
			results.offer(new DefaultProducerResult(index, cause));
			logger.error("Produce failure, topic: {}, QoS: {}", message.topic(), message.qos(), cause);
		}
		else
		{
			if (logger.isDebugEnabled())
			{
				logger.debug("Produce successful, topic: {}, QoS: {}", message.topic(), message.qos());
			}
		}
		latch.countDown();
	}
}
